package com.atguigu.chapter07.d_state;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.*;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.util.Iterator;

/**
 * @ClassName: Flink01_State_Operator_List
 * @Description:
 * @Author: kele
 * @Date: 2021/4/7 19:38
 *
 *来自一个流的一些数据需要广播到所有下游任务，在那里它被本地存储，并用于处理另一个流上的所有传入元素。
 *
 * 一个流控制另一个流
 *
 *
 * 流程：
 *      1、获取两个流
 *      2、将控制流作为广播流
 *      3、将两个流进行connect
 *      4、通过parcess的BroadcastProcessFunction方法进行处理
 *          a) processElement
 *             1、通过ctx先获取广播状态。必须先获取，否则会报空指针异常
 *             2、判断广播状态，进行处理数据
 *          b)processBroadcastElement
 *             1、获取状态，（在同一个程序中，广播状态的参数必须要一致）
 *             2、更新状态
 *
 *
 *
 *
 **/
public class Flink02_State_Operator_Broadcast {

    public static void main(String[] args) {

        Configuration conf = new Configuration();
        conf.setInteger("rest.port",20000);

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(2);


        //1、获取连个流
        DataStreamSource<String> dataStream = env.socketTextStream("hadoop162", 8888);
        DataStreamSource<String> controlStream = env.socketTextStream("hadoop162", 9999);

        MapStateDescriptor<String, String> cstate = new MapStateDescriptor<>("controlState", String.class, String.class);
        //2、将控制流作为广播流
        //下面控制流的名称要和这个一致
        BroadcastStream<String> controlState = controlStream
                .broadcast(cstate);


        dataStream.connect(controlState)
                .process(new BroadcastProcessFunction<String, String, String>() {



                    @Override
                    public void processElement(String value, ReadOnlyContext ctx, Collector<String> out) throws Exception {

                        ReadOnlyBroadcastState<String, String> controlState1 = ctx.getBroadcastState(cstate);

                        String aSwitch = controlState1.get("switch");
                        if("a".equals(aSwitch)){
                            out.collect("a处理方式处理数据："+value);
                        }else if("b".equals(aSwitch)){
                            out.collect("b处理方式处理数据："+value);
                        }else{
                            out.collect("默认处理方式处理：" + value);
                        }

                    }

                    /**
                     * 控制流
                     * @param value
                     * @param ctx
                     * @param out
                     * @throws Exception
                     *
                     * 1、获取广播流的状态
                     * 2、更新广播流中的状态
                     */
                    @Override
                    public void processBroadcastElement(String value, Context ctx, Collector<String> out) throws Exception {

                        //广播流中的状态是以map的形式保存
                        BroadcastState<String, String> controlState = ctx
                                .getBroadcastState(
                                        //*************此处的广播状态要和之前创建时的广播状态一致***************//
                                        cstate);

                        //保障每次只有一个控制状态，所以将key设置为相同的
                        controlState.put("switch",value);

                    }
                })
                .print();


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

}
